package org.apache.flume.source.http;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.SslContextAwareAbstractSource;
import org.apache.flume.tools.FlumeBeanConfigurator;
import org.apache.flume.tools.HTTPServerConstraintUtil;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpConnectionFactory;
import org.eclipse.jetty.server.SecureRequestCustomizer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.jmx.MBeanContainer;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * 通过http post和GET接受Flume事件的源。
 * GET只能用于实验。
 * HTTP请求由一个可插入的“handler”转换为flume事件，该handler必须实现HTTPSourceHandler接口。 此处handler接受HttpServletRequest并返回flume事件列表。
 * 从一个Http请求处理的所有事件都在一个事务中提交到该通channel，从而提高了文件通道等通道的效率。
 * 如果handler抛出异常，则此源将返回HTTP状态400。
 * 如果channel已满，或者源无法向channel追加事件，则源将返回HTTP 503-暂时不可用状态。
 *
 * 在一个post请求中发送的所有事件都被视为一个批处理，并在一个事务中插入到通道中。
 *
 * http source基于Jetty 9.4，并提供设置额外的Jetty特定参数的能力，这些参数将直接传递给Jetty组件。
 */
public class HTTPSource extends SslContextAwareAbstractSource implements EventDrivenSource, Configurable {
  /**
   * 实现http source有两种方法:
   * a. 全局就启动一个http server, 在每个source 里面使用connector绑定到要监听的端口上去，接收http请求。
   *     用一个http server,服务多个source, 去监听多个端口号
   * b. 每一个source都有自己的http server实例，监听这个source的端口号
   *
   * b方法更加高效，因为jetty是高性能的web服务器，不允许把一个servlet绑定到一个connector上去的，所以每个请求都会经过所有的servlet,直到找到对应的servlet为止。
   *
   */
  private static final Logger LOG = LoggerFactory.getLogger(HTTPSource.class);
  private volatile Integer port;
  private volatile Server srv;
  private volatile String host;
  private HTTPSourceHandler handler;
  private SourceCounter sourceCounter;

  private Context sourceContext;

  @Override
  public void configure(Context context) {
    configureSsl(context);
    sourceContext = context;
    try {
      port = context.getInteger(HTTPSourceConfigurationConstants.CONFIG_PORT);
      host = context.getString(HTTPSourceConfigurationConstants.CONFIG_BIND, HTTPSourceConfigurationConstants.DEFAULT_BIND);

      Preconditions.checkState(host != null && !host.isEmpty(), "HTTPSource hostname specified is empty");
      Preconditions.checkNotNull(port, "HTTPSource requires a port number to be specified");

      String handlerClassName = context.getString( HTTPSourceConfigurationConstants.CONFIG_HANDLER, HTTPSourceConfigurationConstants.DEFAULT_HANDLER).trim();

      Class<? extends HTTPSourceHandler> clazz = (Class<? extends HTTPSourceHandler>) Class.forName(handlerClassName);
      handler = clazz.getDeclaredConstructor().newInstance();

      Map<String, String> subProps = context.getSubProperties( HTTPSourceConfigurationConstants.CONFIG_HANDLER_PREFIX);
      handler.configure(new Context(subProps));
    } catch (ClassNotFoundException ex) {
      LOG.error("Error while configuring HTTPSource. Exception follows.", ex);
      Throwables.propagate(ex);
    } catch (ClassCastException ex) {
      LOG.error("Deserializer is not an instance of HTTPSourceHandler." + "Deserializer must implement HTTPSourceHandler.");
      Throwables.propagate(ex);
    } catch (Exception ex) {
      LOG.error("Error configuring HTTPSource!", ex);
      Throwables.propagate(ex);
    }
    if (sourceCounter == null) {
      sourceCounter = new SourceCounter(getName());
    }
  }

  @Override
  public void start() {
    Preconditions.checkState(srv == null, "Running HTTP Server found in source: " + getName() + " before I started one. Will not attempt to start.");
    QueuedThreadPool threadPool = new QueuedThreadPool();
    if (sourceContext.getSubProperties("QueuedThreadPool.").size() > 0) {
      FlumeBeanConfigurator.setConfigurationFields(threadPool, sourceContext);
    }
    // 核心源码， jetty Server
    srv = new Server(threadPool); //

    MBeanContainer mbContainer = new MBeanContainer(ManagementFactory.getPlatformMBeanServer());
    srv.addEventListener(mbContainer);
    srv.addBean(mbContainer);

    HttpConfiguration httpConfiguration = new HttpConfiguration();
    httpConfiguration.addCustomizer(new SecureRequestCustomizer());

    FlumeBeanConfigurator.setConfigurationFields(httpConfiguration, sourceContext);
    // 连接组件 connector
    ServerConnector connector = getSslContextSupplier().get().map(
      sslContext -> {
        SslContextFactory sslCtxFactory = new SslContextFactory();
        sslCtxFactory.setSslContext(sslContext);
        sslCtxFactory.setExcludeProtocols(getExcludeProtocols().toArray(new String[]{}));
        sslCtxFactory.setIncludeProtocols(getIncludeProtocols().toArray(new String[]{}));
        sslCtxFactory.setExcludeCipherSuites(getExcludeCipherSuites().toArray(new String[]{}));
        sslCtxFactory.setIncludeCipherSuites(getIncludeCipherSuites().toArray(new String[]{}));

        FlumeBeanConfigurator.setConfigurationFields(sslCtxFactory, sourceContext);

        httpConfiguration.setSecurePort(port);
        httpConfiguration.setSecureScheme("https");

        return new ServerConnector(srv, new SslConnectionFactory(sslCtxFactory, HttpVersion.HTTP_1_1.asString()), new HttpConnectionFactory(httpConfiguration));
    }).orElse(
        new ServerConnector(srv, new HttpConnectionFactory(httpConfiguration))
    );

    // connector设置 host 和 port
    connector.setPort(port);
    connector.setHost(host);
    connector.setReuseAddress(true);

    FlumeBeanConfigurator.setConfigurationFields(connector, sourceContext);

    // 一个server 可以有多个connect
    srv.addConnector(connector);

    try {

      ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
      context.setContextPath("/");
      srv.setHandler(context);

      // 核心代码：FlumeHTTPServlet
      context.addServlet(new ServletHolder(new FlumeHTTPServlet()),"/"); //
      context.setSecurityHandler(HTTPServerConstraintUtil.enforceConstraints());
      // 启动 jetty Server
      srv.start();

    } catch (Exception ex) {
      LOG.error("Error while starting HTTPSource. Exception follows.", ex);
      Throwables.propagate(ex);
    }
    Preconditions.checkArgument(srv.isRunning());
    sourceCounter.start();
    super.start();
  }

  @Override
  public void stop() {
    try {
      srv.stop();
      srv.join();
      srv = null;
    } catch (Exception ex) {
      LOG.error("Error while stopping HTTPSource. Exception follows.", ex);
    }
    sourceCounter.stop();
    LOG.info("Http source {} stopped. Metrics: {}", getName(), sourceCounter);
  }

  private class FlumeHTTPServlet extends HttpServlet {

    private static final long serialVersionUID = 4891924863218790344L;

    @Override
    public void doPost(HttpServletRequest request, HttpServletResponse response) throws IOException {
      List<Event> events = Collections.emptyList(); //create empty list
      try {
        // 核心方法，在这里拿到一堆events，每一个event都是带着headers的一个JSON串，对应的就是一条用户行为日志
        events = handler.getEvents(request); // JSONHandler
      } catch (HTTPBadRequestException ex) {
        LOG.warn("Received bad request from client. ", ex);
        sourceCounter.incrementEventReadFail();
        response.sendError(HttpServletResponse.SC_BAD_REQUEST, "Bad request from client. " + ex.getMessage());
        return;
      } catch (Exception ex) {
        LOG.warn("Deserializer threw unexpected exception. ", ex);
        sourceCounter.incrementEventReadFail();
        response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Deserializer threw unexpected exception. " + ex.getMessage());
        return;
      }

      sourceCounter.incrementAppendBatchReceivedCount();
      sourceCounter.addToEventReceivedCount(events.size());

      try {
        // 核心方法，getChannelProcessor()，Event处理器
        getChannelProcessor().processEventBatch(events); //
      } catch (ChannelException ex) {
        LOG.warn("Error appending event to channel. Channel might be full. Consider increasing the channel  capacity or make sure the sinks perform faster.", ex);
        sourceCounter.incrementChannelWriteFail();
        response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Error appending event to channel. Channel might be full." + ex.getMessage());
        return;
      } catch (Exception ex) {
        LOG.warn("Unexpected error appending event to channel. ", ex);
        sourceCounter.incrementGenericProcessingFail();
        response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Unexpected error while appending event to channel. " + ex.getMessage());
        return;
      }

      response.setCharacterEncoding(request.getCharacterEncoding());
      response.setStatus(HttpServletResponse.SC_OK);
      response.flushBuffer();
      sourceCounter.incrementAppendBatchAcceptedCount();
      sourceCounter.addToEventAcceptedCount(events.size());
    }

    @Override
    public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
      doPost(request, response);
    }
  }

  @Override
  protected void configureSsl(Context context) {
    handleDeprecatedParameter(context, "ssl", "enableSSL");
    handleDeprecatedParameter(context, "exclude-protocols", "excludeProtocols");
    handleDeprecatedParameter(context, "keystore-password", "keystorePassword");

    super.configureSsl(context);
  }

  private void handleDeprecatedParameter(Context context, String newParam, String oldParam) {
    if (!context.containsKey(newParam) && context.containsKey(oldParam)) {
      context.put(newParam, context.getString(oldParam));
    }
  }

}
